package com.kilomob.grady.client.netty;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.HashedWheelTimer;

import java.net.InetSocketAddress;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.kilomob.grady.client.Client;
import com.kilomob.grady.codec.protostuff.ProtostuffCodec;
import com.kilomob.grady.codec.protostuff.ProtostuffDecoder;
import com.kilomob.grady.codec.protostuff.ProtostuffEncoder;
import com.kilomob.grady.concurrent.DefaultThreadFactory;
import com.kilomob.grady.handler.netty.ConnectionWatchdog;
import com.kilomob.grady.handler.netty.ConnectorIdleStateTrigger;
import com.kilomob.grady.handler.netty.MessageSendHandler;

/**
 * 
 * NettyClient
 * 
 * @author zhouzhixiang
 * @date 2017年3月15日
 * @since 1.0
 */
public class NettyClient implements Client {
    private static Logger LOGGER = LoggerFactory.getLogger(NettyClient.class);

    private EventLoopGroup eventLoopGroup = new NioEventLoopGroup(Runtime.getRuntime()
            .availableProcessors() * 2 + 1, new ThreadFactoryBuilder().setNameFormat(
            "Grady-client-%s").build());
    
    // handlers
    private final ConnectorIdleStateTrigger idleStateTrigger = new ConnectorIdleStateTrigger();
    protected final HashedWheelTimer timer = new HashedWheelTimer(new DefaultThreadFactory("connector.timer"));
    /** 初始连接数 **/
    private int connections = 1;
    /** 客户端名称 **/
    private String name;

    /** 客户端Id **/
    private String id;

    /** 连接超时时间 **/
    private int timeout;
    
    private boolean reconnect = true;

    public NettyClient() {}

    public void connect(final String host, final int port, final String protocol)
            throws InterruptedException {
        if (!isConnected()) {
            Bootstrap boot = new Bootstrap();
            boot.group(eventLoopGroup)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS,
                            timeout == 0 ? 10 * 1000 : timeout)
                    .option(ChannelOption.SO_KEEPALIVE, true);

            NettyChannelFactory channelFactory = new NettyChannelFactory(id);
            channelFactory.setConnections(connections);
            boot.channelFactory(channelFactory);

            final String serverId = id;
            final InetSocketAddress serverAddress = new InetSocketAddress(host, port);
            final ProtostuffCodec util = new ProtostuffCodec();
            util.setRpcDirect(false);
            final ConnectionWatchdog watchdog = new ConnectionWatchdog(boot, timer, serverAddress) {

                public ChannelHandler[] handlers() {
                    return new ChannelHandler[] {
                            this,
                            new IdleStateHandler(0, 4, 0, TimeUnit.SECONDS),
                            idleStateTrigger,
                            new ProtostuffEncoder(util),
                            new ProtostuffDecoder(util),
                            new MessageSendHandler(serverId)
                    };
                }
            };
            
            boot.handler(new ChannelInitializer<Channel>() {

                @Override
                protected void initChannel(Channel ch) throws Exception {
                    ch.pipeline().addLast(watchdog.handlers());
                }
            });

            for (int i = 0; i < connections; i++) {
                final int j = i + 1;
                ChannelFuture channelFuture = boot.connect(host, port);

                channelFuture.awaitUninterruptibly();
                channelFuture.isDone();

                if (channelFuture.isCancelled()) {
                    LOGGER.warn("someone maybe cancelled connecting to server!");
                } else if (!channelFuture.isSuccess()) {
                    LOGGER.error("connected to server error:", channelFuture.cause());
                    if(reconnect){
                        timer.newTimeout(watchdog, timeout, TimeUnit.MILLISECONDS);
                    }
                } else {
                    LOGGER.info("{}-client-{} connected the server {}:{} success", name == null
                            ? "Grady"
                            : name, j, host, port);
//                    MessageSendHandler handler =
//                            channelFuture.channel().pipeline().get(MessageSendHandler.class);
//                    RpcServerLoader.getInstance().setMessageSendHandler(handler);
                }

            }
        }
    }
    

    public boolean isReconnect() {
        return reconnect;
    }

    public void setReconnect(boolean reconnect) {
        this.reconnect = reconnect;
    }

    public boolean isConnected() {
        return false;
    }


    public int getConnections() {
        return connections;
    }

    public void setConnections(int connections) {
        this.connections = connections;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public int getTimeout() {
        return timeout;
    }

    public void setTimeout(int timeout) {
        this.timeout = timeout;
    }


    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

    public void stop() {
        eventLoopGroup.shutdownGracefully();
    }

}
